package cn.cmft.jc.sink;

import com.alibaba.fastjson2.JSONArray;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.SimpleStringSerializer;

import java.util.List;
import java.util.Properties;

public class WithDorisSink {

    private final DorisSink.Builder<String> dorisSinkBuilder;
    private final DorisOptions.Builder dorisOptionsBuilder;
    private final DorisExecutionOptions.Builder dorisExectionOptionsBuilder;

    public WithDorisSink() {
        this.dorisSinkBuilder = DorisSink.builder();
        this.dorisOptionsBuilder = DorisOptions.builder();
        this.dorisExectionOptionsBuilder = DorisExecutionOptions.builder();
        init();
    }

    private void init(){
       dorisOptionsBuilder.setFenodes("192.168.17.137:8030")
               .setTableIdentifier("pay_bigdata.table_struct_change_impact_log")
               .setUsername("root")
               .setPassword("123456");

        Properties properties = new Properties();
        properties.setProperty("format","json");
        properties.setProperty("read_json_by_line","true");
        properties.setProperty("sink.batch.size","1");
        properties.setProperty("sink.batch.interval","1");

        dorisExectionOptionsBuilder
                .setLabelPrefix("mysql-doris3660")
                .setDeletable(false)
                .setStreamLoadProp(properties);

        dorisSinkBuilder.setDorisReadOptions(DorisReadOptions.builder().build())
                .setDorisExecutionOptions(dorisExectionOptionsBuilder.build())
                .setSerializer(new SimpleStringSerializer())
                .setDorisOptions(dorisOptionsBuilder.build());
    }
    public DorisSink<String>  build(){
        return this.dorisSinkBuilder.<String>build();
    }
}
